package org.jxy.study.rpc.netty.hander;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import org.jxy.study.rpc.message.DataMsg;
import org.jxy.study.rpc.message.Request;
import org.jxy.study.rpc.message.Response;
import org.jxy.study.rpc.util.JsonUtil;
import org.springframework.stereotype.Component;

/**
 * @author : porteryao
 * @description : 客户端响应处理器
 * @date : 2021/3/12 14:20
 */
@Sharable
@Component
public class NettyClientResponseHandler extends SimpleChannelInboundHandler<Response> {

    /**
     * 等待响应队列
     */
    private Map<String, BlockingQueue<Response>> questMap = new ConcurrentHashMap<>();
    /**
    *将返回结果放在阻塞队列
    */
    public BlockingQueue<Response> sendMsg(Channel channel, Request request) {
        return getResponses(channel, request, questMap);
    }

    static BlockingQueue<Response> getResponses(Channel channel, Request request,
        Map<String, BlockingQueue<Response>> questMap) {
        BlockingQueue<Response> blockingQueue = new LinkedBlockingQueue<>();
        questMap.put(request.getId(), blockingQueue);
        byte[] bytes = JsonUtil.objectToJsonByteArray(request);
        DataMsg dataMsg = new DataMsg(bytes.length, bytes);
        channel.writeAndFlush(dataMsg);
        return blockingQueue;
    }


    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Response response) {
        BlockingQueue<Response> responseBlockingQueue = questMap.get(response.getRequestId());
        responseBlockingQueue.add(response);
        questMap.remove(response.getRequestId());
    }
}
